package netty.hanlder;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import server.KoalasServerPublisher;

import java.util.List;

/**
 * Copyright (C) 2018
 * All rights reserved
 * User: yulong.zhang
 * Date:2018年11月23日11:13:33
 */
public class KoalasDecoder extends ByteToMessageDecoder {

    private final static Logger logger = LoggerFactory.getLogger(KoalasDecoder.class);

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 通过读取四个字节的长度来决定消息体长度，然后根据消息体长度来读取所有的字节流数据。
        // decodeFrameSize方法将四个字节流转成int类型。
        try {
            // 通过读取四个字节的长度来决定消息体长度，这是安照thrift的协议的定义的
            // 这里小于4个字节的长度表示不是一个完整的消息
            if (in.readableBytes() < 4) {
                return;
            }

            // 标记buffer中可读索引readerIndex
            in.markReaderIndex();
            byte[] b = new byte[4];
            // 从当前readerIndex开始将此缓冲区的数据传输到指定的目标，并将readerIndex增加传输的字节数
            // 读取前4个字节
            in.readBytes(b);

            // 将4个字节流转换为int类型，获取消息体长度
            int length = decodeFrameSize(b);

            // in.readableBytes()当前可读字节大小
            // 若当前可读字节数小于实际长度，说明消息不完整则直接返回，直到当前可以获取到的字节数等于实际长度
            if (in.readableBytes() < length) {
                //重置可读索引，这里一定要重置索引，不然会导致下次完整消息过来时无法读取到前4个标识消息长度的字节
                in.resetReaderIndex();
                return;
            }
            // 将当前readerIndex重新定位到此缓冲区中标记的readerIndex，因为在上面的readBytes中已经读了4个字节了
            in.resetReaderIndex();
            // 将in中可读的数据复制到一个新的buf中
            ByteBuf fream = in.readRetainedSlice(4 + length);
            in.resetReaderIndex();

            // skipBytes(int length)该方法实际上就是让readerIndex += length
            // 这里这样做的意思个人理解为，有多个消息在这个buf中，当本条消息读完了后，
            // 跳过已经处理过的消息的所有字节
            in.skipBytes(4 + length);
            out.add(fream);
        } catch (Exception e) {
            logger.error("decode error", e);
        }

    }

    public static final int decodeFrameSize(byte[] buf) {
        return (buf[0] & 255) << 24 | (buf[1] & 255) << 16 | (buf[2] & 255) << 8 | buf[3] & 255;
    }
}
